package com.hnichr.emqx;

import com.hnichr.stop.message.MessageArrivedHandle;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Iterator;
import java.util.List;

@Configuration
public class MqttConfig {

    @Autowired
    private List<MessageArrivedHandle> handleList;

    private static final Logger logger = LoggerFactory.getLogger(MqttConfig.class);

    @Autowired
    private MqttProperties properties;

    @Bean
    public MqttClient initClient() throws Exception{
        //新建mqtt连接
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(properties.getUsername());
        options.setPassword(properties.getPassword().toCharArray());
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(5000);
        //新建mqtt客户端
        MqttClient client = new MqttClient(properties.getUrl(), properties.getClientId());
        client.connect(options);
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                logger.error(">>>>>>>>>>>>>>>>>>>>>>>>>>>连接丢失");
            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage){
                if(mqttMessage==null){
                    return;
                }
                String message = new String(mqttMessage.getPayload());
                try{
                    Iterator<MessageArrivedHandle> it = handleList.iterator();
                    while (it.hasNext()){
                        MessageArrivedHandle handle = it.next();
                        if(handle.support(topic,message)){
                            handle.handle(topic,message);
                        }
                    }
                }
                catch (Exception e){
                    logger.error(">>>>>>>>>>>>>>>>>>>>>>>>>>>emqx消息处理异常",e);
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
            }
        });
        return client;
    }
}
